package com.xrui.hbase.filter;

import com.google.common.base.Objects;
import com.google.common.base.Preconditions;

import com.xrui.hbase.DataRequest;
import org.apache.commons.lang.builder.HashCodeBuilder;
import org.apache.hadoop.hbase.filter.Filter;
import org.codehaus.jackson.JsonNode;
import org.codehaus.jackson.node.JsonNodeFactory;
import org.codehaus.jackson.node.ObjectNode;

import java.io.IOException;


public final class ColumnPrefixFilter extends RowFilter {

    /**
     * All StripValueRowFilter instances are the same: generate hash-code ahead of time.
     */
    private static final int HASH_CODE =
        new HashCodeBuilder().append(ColumnPrefixFilter.class).toHashCode();
    private byte[] mPrefix;

    public ColumnPrefixFilter(byte[] rowKeyPrefix) {
        Preconditions.checkState(rowKeyPrefix != null, "ColumnPrefixFilter should have a prefix.");
        mPrefix = rowKeyPrefix;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public DataRequest getDataRequest() {
        return DataRequest.builder().build();
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public Filter toHBaseFilter() throws IOException {
        return new org.apache.hadoop.hbase.filter.ColumnPrefixFilter(mPrefix);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean equals(Object other) {
        if (!(other instanceof ColumnPrefixFilter)) {
            return false;
        } else {
            final ColumnPrefixFilter otherFilter = (ColumnPrefixFilter) other;
            return Objects.equal(otherFilter.mPrefix, mPrefix);
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public int hashCode() {
        return HASH_CODE;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public String toString() {
        return Objects.toStringHelper(ColumnPrefixFilter.class).toString();
    }

    /**
     * {@inheritDoc}
     */
    @Override
    protected JsonNode toJsonNode() {
        final ObjectNode root = JsonNodeFactory.instance.objectNode();
        root.put("prefix", mPrefix);
        return root;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    protected Class<? extends ColumnPrefixFilterDeserializer> getDeserializerClass() {
        return ColumnPrefixFilterDeserializer.class;
    }

    /**
     * Deserializes {@code HasColumnDataRowFilter}.
     */
    public static final class ColumnPrefixFilterDeserializer
        implements RowFilterDeserializer {
        /**
         * {@inheritDoc}
         */
        @Override
        public RowFilter createFromJson(JsonNode root) {
            final String prefix = root.path("prefix").getTextValue();
            return new ColumnPrefixFilter(prefix.getBytes());
        }
    }

}
